package flink_p1

import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.{KeyedStream, StreamExecutionEnvironment, createTypeInformation}
import org.apache.flink.util.Collector

/**
 * 通过process直接使用更底层的API
 */
object FlinkTest_10_Process {

  /**
   * 统计车速，对于超过100的车速在2秒之后发出警告
   */
  case class CarInfo(carId: String, speed: Long)

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val stream = env.socketTextStream("127.0.0.1", 8889)

    val keyedStream: KeyedStream[CarInfo, String] = stream.map(data => {
      val splits = data.split(" ")
      val carId = splits(0)
      val speed = splits(1).toLong
      CarInfo(carId, speed)
    }).keyBy(_.carId)


    // KeyedStream调用process需要传入KeyedProcessFunction
    // DataStream调用process需要传入ProcessFunction
    keyedStream.process(new KeyedProcessFunction[String, CarInfo, String] {

      override def processElement(value: CarInfo, ctx: KeyedProcessFunction[String, CarInfo, String]#Context, out: Collector[String]): Unit = {

        if (value.speed > 100) {
          val currentTime = ctx.timerService().currentProcessingTime()
          val timerTime = currentTime + 2 * 1000
          ctx.timerService().registerProcessingTimeTimer(timerTime)
        }
      }

      override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, CarInfo, String]#OnTimerContext, out: Collector[String]): Unit = {
        var warnMsg = "warn... time:" + timestamp + " carID:" + ctx.getCurrentKey
        out.collect(warnMsg)
      }
    }).print()


    env.execute()
  }
}
